/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package elasticsearch

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"github.com/olivere/elastic/v7"
	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
	"math"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	FieldPayload string = "_payload"

	configKeyAddresses  = "elasticSearch.addresses"
	configKeyUsername   = "elasticSearch.username"
	configKeyPassword   = "elasticSearch.password"
	configKeyMaxRetries = "elasticSearch.maxRetries"
	configKeySniff      = "elasticSearch.sniff"
	configKeyRetryLimit = "store.retryLimit"

	// 若多个地址,逗号分隔
	configKeyEsAddresses  = "addresses"
	configKeyEsUsername   = "username"
	configKeyEsPassword   = "password"
	configKeyEsMaxRetries = "maxRetries"
	configKeyEsSniff      = "sniff"
)

var client *Client

var once sync.Once
var log logging.Log
var RetryLimit int

// deprecated
func GetClient() *Client {
	once.Do(func() {
		c := config.ElasticSearchConfig{
			Addresses:  config.GetStringConfigValues(configKeyAddresses, []string{"localhost:9200"}),
			Username:   config.GetStringConfigValue(configKeyUsername, ""),
			Password:   config.GetStringConfigValue(configKeyPassword, ""),
			MaxRetries: config.GetIntConfigValue(configKeyMaxRetries, 3),
			Sniff:      config.GetBoolConfigValue(configKeySniff, false),
		}

		var err error
		client, err = NewClient(c)
		if err != nil {
			panic(err)
		}
		RetryLimit = config.GetIntConfigValue(configKeyRetryLimit, math.MaxInt32)
	})
	return client
}

// NewClient 新建一个ES客户端
func NewClient(config config.ElasticSearchConfig) (*Client, error) {
	var opts []elastic.ClientOptionFunc
	opts = append(opts, elastic.SetURL(config.Addresses...))
	if config.Username != "" {
		opts = append(opts, elastic.SetBasicAuth(config.Username, config.Password))
	}
	opts = append(opts, elastic.SetSniff(config.Sniff))
	if config.MaxRetries <= 0 {
		opts = append(opts, elastic.SetRetrier(elastic.NewStopRetrier()))
	} else {
		//重试间隔从100ms开始，以2为底指数级增长
		ticks := make([]int, config.MaxRetries)
		for i := 0; i < len(ticks); i++ {
			ticks[i] = 100*2 ^ (i - 1)
		}
		backoff := elastic.NewSimpleBackoff(ticks...)
		retrier := elastic.NewBackoffRetrier(backoff)
		opts = append(opts, elastic.SetRetrier(retrier))
	}

	realClient, err := elastic.NewClient(opts...)

	if err != nil {
		return nil, err
	}

	// todo
	log = logging.GetLogWithCommonConf(logging.ConfigKeyElasticSearchLogLevel, logging.LogFieldEs, logging.LogFieldEs)

	client = &Client{realClient}
	return client, nil
}

func Map2EsConfig(parameters map[string]string) (config.ElasticSearchConfig, error) {
	esConfig := config.ElasticSearchConfig{}

	if parameters == nil {
		return esConfig, types.ErrNilEsConfig
	}

	if addresses, ok := parameters[configKeyEsAddresses]; ok {
		esConfig.Addresses = strings.Split(addresses, ",")
	}
	if username, ok := parameters[configKeyEsUsername]; ok {
		esConfig.Username = username
	}
	if password, ok := parameters[configKeyEsPassword]; ok {
		esConfig.Password = password
	}
	if maxLimit, ok := parameters[configKeyEsMaxRetries]; ok {
		if esConfig.MaxRetries, _ = strconv.Atoi(maxLimit); esConfig.MaxRetries < 3 {
			esConfig.MaxRetries = 3
		}
	} else {
		esConfig.MaxRetries = 3
	}
	if sniffCfg, ok := parameters[configKeyEsSniff]; ok {
		if enableSniff, err := strconv.ParseBool(sniffCfg); err == nil {
			esConfig.Sniff = enableSniff
		}
	}

	return esConfig, nil
}

// GetRequest get请求
type GetRequest struct {
	Index string
	DocID string
}

// SetRequest set请求
type SetRequest struct {
	Index string
	DocID string
	Value []byte
}

func value2DocBody(value []byte) map[string]interface{} {
	docBody := make(map[string]interface{})
	docBody[FieldPayload] = value
	return docBody
}

func docBody2Value(source []byte) ([]byte, error) {
	var docBody map[string]interface{}
	if err := json.Unmarshal(source, &docBody); err != nil {
		return nil, err
	}
	if payload, ok := docBody[FieldPayload]; ok {
		if payloadString, ok := payload.(string); ok {
			if value, err := base64.StdEncoding.DecodeString(payloadString); err == nil {
				return value, nil
			}
		}
	}
	return nil, types.ErrWrongDocBody
}

// Set 在指定索引下保存kv
func (cli *Client) Set(index, key string, value []byte) error {

	if index == "" {
		return types.ErrEmptyIndexName
	}

	if key == "" {
		return types.ErrEmptyDocID
	}

	// allow bodyString to be empty

	defer log.ElapsedTime("Set", time.Now())

	_, err := cli.realClient.Index().
		Index(index).
		Id(key).
		BodyJson(value2DocBody(value)).
		Refresh("true").
		Do(context.Background())

	return err
}

// Has 判断指定索引下key是否存在
func (cli *Client) Has(index, key string) (bool, error) {
	if index == "" {
		return false, types.ErrEmptyIndexName
	}

	if key == "" {
		return false, types.ErrEmptyDocID
	}

	defer log.ElapsedTime("Has", time.Now())

	_, err := cli.realClient.Get().Index(index).Id(key).Do(context.Background())
	if err == nil {
		return true, nil
	}
	if isNotFoundError(err) {
		return false, nil
	}
	return false, err
}

func (cli *Client) Delete(index, key string) error {
	if index == "" {
		return types.ErrEmptyIndexName
	}

	if key == "" {
		return types.ErrEmptyDocID
	}

	defer log.ElapsedTime("Delete", time.Now())

	_, err := cli.realClient.Delete().
		Index(index).
		Id(key).
		Refresh("true").
		Do(context.Background())
	if err == nil || isNotFoundError(err) {
		return nil
	}
	return err
}

func (cli *Client) Get(index, key string) ([]byte, error) {
	if index == "" {
		return nil, types.ErrEmptyIndexName
	}

	if key == "" {
		return nil, types.ErrEmptyDocID
	}

	defer log.ElapsedTime("Get", time.Now())

	rs, err := cli.realClient.Get().Index(index).Id(key).Do(context.Background())
	if err != nil {
		return nil, err
	}

	return docBody2Value(rs.Source)
}

// 批量get 入参不做去重,仅当查询出错才返回错误；返回数组允许元素为nil
func (cli *Client) Gets(requests []GetRequest) ([][]byte, error) {
	if len(requests) == 0 {
		return nil, types.ErrNilRequests
	}

	defer log.ElapsedTime("Gets", time.Now())

	items := make([]*elastic.MultiGetItem, len(requests))
	for i, request := range requests {
		if request.Index == "" {
			return nil, types.ErrEmptyIndexName
		}
		if request.DocID == "" {
			return nil, types.ErrEmptyDocID
		}
		items[i] = elastic.NewMultiGetItem().Index(request.Index).Id(request.DocID)
	}

	resp, err := cli.realClient.Mget().Add(items...).Do(context.Background())
	if err != nil {
		return nil, err
	}

	results := make([][]byte, len(requests))
	for i, doc := range resp.Docs {
		if doc.Found {
			if value, err := docBody2Value(doc.Source); err == nil {
				results[i] = value
			}
		}
	}
	return results, nil
}

// 批量set 若仅部分写入成功，则仍报错，由调用方负责重新发起写入
func (cli *Client) Sets(requests []SetRequest) error {
	if len(requests) == 0 {
		return types.ErrNilRequests
	}

	defer log.ElapsedTime("Gets", time.Now())

	items := make([]elastic.BulkableRequest, len(requests))
	for i, request := range requests {
		if request.Index == "" {
			return types.ErrEmptyIndexName
		}
		if request.DocID == "" {
			return types.ErrEmptyDocID
		}
		items[i] = elastic.NewBulkIndexRequest().Index(request.Index).Id(request.DocID).Doc(value2DocBody(request.Value))
	}

	resp, err := cli.realClient.Bulk().Refresh("true").Add(items...).Do(context.Background())
	if err != nil {
		return err
	}

	if len(resp.Failed()) > 0 {
		return types.ErrFailedResult
	}

	return nil
}
